package top.v5it.japi.ddd.example.points.event;

import cn.hutool.core.util.NumberUtil;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
import top.v5it.japi.ddd.example.domain.event.ProductReviewDTO;
import top.v5it.japi.ddd.example.points.entity.PointsRecord;
import top.v5it.japi.ddd.example.domain.event.OrderDTO;
import top.v5it.japi.ddd.example.points.service.PointsRecordService;

/**
 * @author zhanpu
 * @date 2023/11/17
 */
@Slf4j
@Component
@RequiredArgsConstructor
public class EventListener {

    public static final String TOPIC_ORDER_NAME = "example_order_event_topic";
    public static final String TOPIC_PRODUCT_REVIEW_NAME = "example_product_review_event_topic";
    private final PointsRecordService pointsRecordService;

    @KafkaListener(topics = TOPIC_ORDER_NAME)
    public void orderEvent(ConsumerRecord<String, OrderDTO> record) {
        final OrderDTO value = record.value();
        log.info("收到消息：{}", value);
        final PointsRecord entity = new PointsRecord();
        final int points = NumberUtil.div(value.getTotal_amount(), 10).intValue();
        entity.setMember_id(value.getMember_id());
        entity.setPoints(points);
        entity.setRecord_type("购买产品");
        pointsRecordService.add(entity);
        log.info("成功新增积分记录");
    }

    @KafkaListener(topics = TOPIC_PRODUCT_REVIEW_NAME)
    public void productReviewEvent(ConsumerRecord<String, ProductReviewDTO> record) {
        final ProductReviewDTO value = record.value();
        log.info("收到消息：{}", value);
        final PointsRecord entity = new PointsRecord();
        final int points = 5;
        entity.setMember_id(value.getMember_id());
        entity.setPoints(points);
        entity.setRecord_type("评价产品");
        pointsRecordService.add(entity);
        log.info("成功新增积分记录");
    }
}
